#include <librdkafka/rdkafkacpp.h>

#include <iostream>
#include <string>
#include <memory>
class DeliveryReportCb : public RdKafka::DeliveryReportCb
{
   public:
    void dr_cb(RdKafka::Message &message) override
    {
        if (message.err())
        {
            std::cerr << "Delivery failed: " << message.errstr() << std::endl;
        }
        else
        {
            std::cout << "Message delivered to " << message.topic_name() << " [" << message.partition() << "] at offset " << message.offset()
                      << std::endl;
        }
    }
};

int main()
{
    std::string brokers = "localhost:9092";
    std::string topic_name = "test_topic";
    std::string errstr;

    // 创建全局配置对象
    std::unique_ptr<RdKafka::Conf> conf(RdKafka::Conf::create(RdKafka::Conf::CONF_GLOBAL));
    if (!conf)
    {
        std::cerr << "Failed to create global config" << std::endl;
        return 1;
    }

    // 创建主题配置对象
    std::unique_ptr<RdKafka::Conf> tconf(RdKafka::Conf::create(RdKafka::Conf::CONF_TOPIC));
    if (!tconf)
    {
        std::cerr << "Failed to create topic config" << std::endl;
        return 1;
    }

    // 设置 broker 列表
    if (conf->set("metadata.broker.list", brokers, errstr) != RdKafka::Conf::CONF_OK)
    {
        std::cerr << "Failed to set broker list: " << errstr << std::endl;
        return 1;
    }

    // 设置交付报告回调
    DeliveryReportCb delivery_report_cb;
    if (conf->set("dr_cb", &delivery_report_cb, errstr) != RdKafka::Conf::CONF_OK)
    {
        std::cerr << "Failed to set delivery report callback: " << errstr << std::endl;
        return 1;
    }

    // 创建生产者
    std::unique_ptr<RdKafka::Producer> producer(RdKafka::Producer::create(conf.get(), errstr));
    if (!producer)
    {
        std::cerr << "Failed to create producer: " << errstr << std::endl;
        return 1;
    }

    // 创建主题对象
    std::unique_ptr<RdKafka::Topic> topic(RdKafka::Topic::create(producer.get(), topic_name, tconf.get(), errstr));
    if (!topic)
    {
        std::cerr << "Failed to create topic: " << errstr << std::endl;
        return 1;
    }

    // 生产消息
    std::string message = "Hello, Kafka!";
    RdKafka::ErrorCode resp = producer->produce(topic.get(), RdKafka::Topic::PARTITION_UA, RdKafka::Producer::RK_MSG_COPY,
                                                const_cast<char *>(message.c_str()), message.size(), nullptr, nullptr);
    if (resp != RdKafka::ERR_NO_ERROR)
    {
        std::cerr << "Produce failed: " << RdKafka::err2str(resp) << std::endl;
    }

    // 刷新生产者，确保所有消息都已发送
    producer->flush(10000);

    return 0;
}
